package com.example.mq_sm.mqserver.core;

import com.example.mq_sm.common.Consumer;
import com.example.mq_sm.common.ConsumerEnv;
import com.example.mq_sm.common.MQException;
import com.example.mq_sm.mqserver.VirtualHost;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * 通过这个类来实现消费消息的核心逻辑
 */
public class ConsumerManager {
    // 持有上层 VirtualHost 的引用，用来操作数据
    private VirtualHost parent;
    // 指定一个线程池，负责去执行具体的回调任务
    private ExecutorService workerPool = Executors.newFixedThreadPool(4);
    // 存放令牌的队列
    private BlockingQueue<String> tokenQueue = new LinkedBlockingDeque<>();
    // 扫描线程
    private Thread scannerThread = null;

    public ConsumerManager(VirtualHost p){
        parent = p;
        scannerThread = new Thread(() ->{
            while(true){
                try{
                    // 拿到令牌
                    String queueName = tokenQueue.take();
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                    if (queue == null){
                        throw new MQException("");
                    }
                    // 从这个队列中消费一个消息
                    synchronized (queue){
                        consumeMessage(queue);
                    }
                } catch (InterruptedException | MQException e){
                    e.printStackTrace();
                }
            }
        });
        // 把线程设为后台线程
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MQException {
        // 找到对应的队列
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if (queue == null){
            throw new MQException("");
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
        synchronized (queue){
            queue.addConsumerEnv(consumerEnv);
            // 如果当前队列已经有一些消息了 就需要立刻消费掉
            int n = parent.getMemoryDataCenter().getMessageCount(queueName);
            for ( int i=0;i<n;i++ ){
                consumeMessage(queue);
            }
        }

    }

    private void consumeMessage(MSGQueue queue){
        // 按照轮询方式，找一个消费者出来
        ConsumerEnv luckyDog = queue.chooseConsumer();
        if (luckyDog == null){
            // 说明当前队列没有消费者
            return;
        }
        // 从队列中取出一个消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if (message == null){
            return;
        }
        // 把消息代入到消费者的回调方法中，丢给线程池执行
        workerPool.submit(() ->{
            try {
                // 把消息放到待确认的集合中
                // 按照逻辑上的需要，为了保证不遗漏这个操作必须放在回调之前
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
                // 执行回调
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(), message.getBasicProperties(), message.getBody());
                // 如果是 自动应答 ，就可以直接进行删除
                // 如果是 手动应答 ，先不处理，交给 basicAck 进行处理
                if (luckyDog.isAutoAck()){
                    if (message.getDeliverMode() == 2){
                        parent.getDiskDataCenter().deleteMessage(queue, message);
                    }
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(), message.getMessageId());
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                    System.out.println();
                }
            } catch (Exception e){
                e.printStackTrace();
            }
        });
    }

}
